package com.lightboat.system.queue;

import cn.hutool.core.util.ObjUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson2.JSON;
import lombok.Setter;
import org.redisson.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * 定时分布式批量处理器
 */
public class DistributedBatchProcessor<T> {
    Logger log = LoggerFactory.getLogger(DistributedBatchProcessor.class);
    private static final String LOCK_NAME = "dbp:lock:";
    private static final String QUEUE_NAME = "dbp:queue:";
    private static final String LAST_PROCESS_KEY = "dbp:time:";
    private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();


    private RedissonClient redisson;
    private RBlockingQueue<T> blockingQueue;
    private Consumer<List<T>> consumer;

    private String threadName = RandomUtil.randomString(6);

    private  int batchSize;

    private  long waitTime , leaseTime, delay = 1000L;
    private TimeUnit unit;

    /**
     * -- SETTER --
     *  设置处理间隔 单位秒
     *
     * @param processInterval
     */
    @Setter
    private  int processInterval = 3; // 秒

    private DistributedBatchProcessor() { }

    public static <T> DistributedBatchProcessor<T> create(RedissonClient redisson, String key, String threadName, int batchSize) {
        Assert.isTrue(batchSize > 0, "处理数需大于0");
        Assert.isTrue(ObjUtil.isNotEmpty(key), "订阅队列不能为空");
        Assert.isTrue(ObjUtil.isNotEmpty(threadName), "线程名称不能为空");
        DistributedBatchProcessor<T> dbp = new DistributedBatchProcessor<T>();
        dbp.blockingQueue = redisson.getBlockingQueue(QUEUE_NAME + key + "_" + SpringUtil.getActiveProfile());
        dbp.redisson = redisson;
        dbp.batchSize = batchSize;
        dbp.threadName = threadName;

        return dbp;
    }

    /**
     * 消费者
     * @param consumer
     */
    public void consumer(Consumer<List<T>> consumer) {
        this.consumer = consumer;
    }

    /**
     * 启动
     * @param waitTime 等待时间
     * @param leaseTime  锁持有时间
     * @param unit 时间单位
     */
    public void setLock(long waitTime, long leaseTime, TimeUnit unit) {
        this.waitTime = waitTime;
        this.leaseTime = leaseTime;
        this.unit = unit;
    }

    /**
     * 设置延迟时间
     * @param delay 延迟时间 单位毫秒
     */
    public void setDelayTime(long delay){
        this.delay = delay;
    }

    public void start() {
        executorService.scheduleWithFixedDelay(() -> {
            // 指定线程名称
            Thread.currentThread().setName(threadName);
            log.debug("线程：{}，队列数: {}", threadName, blockingQueue.size());
            RLock lock = redisson.getLock(LOCK_NAME + threadName);
            try {
                // 尝试获取分布式锁（等待最多500ms，锁持有时间4秒）
                if (lock.tryLock(waitTime, leaseTime, unit)) {
                    RAtomicLong lastProcess = redisson.getAtomicLong(LAST_PROCESS_KEY + threadName);
                    long currentTime = System.currentTimeMillis();

                    // 验证时间间隔
                    if (currentTime - lastProcess.get() >= processInterval * 1000L) {
                        processBatch();
                        lastProcess.set(currentTime);
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        }, 0, delay, TimeUnit.MILLISECONDS); // 检查间隔500ms
    }

    private void processBatch() {
        List<T> messages = new ArrayList<>(batchSize);
        
        try {
            // 批量获取消息（非阻塞方式）
            for (int i = 0; i < batchSize; i++) {
                T poll = blockingQueue.poll();// 立即返回，不阻塞
                if (poll == null) break;
                messages.add(poll);
            }

            if (!messages.isEmpty()) {
                // 业务处理逻辑
                consumer.accept(messages);
            }
        } catch (Exception e) {
            // 处理失败时回滚消息
            log.error("Processing failed", e);
            log.error("message:{}", JSON.toJSONString(messages));
        }
    }

    public boolean publishAsync(T task) {
        if (null == consumer) {
            log.error("消费者不存在，请先配置消费者后启动!");
            return false;
        }
        RFuture<Boolean> future = blockingQueue.offerAsync(task);
        return future.isDone();
    }

    public void shutdown() {
        executorService.shutdown();
        redisson.shutdown();
    }
}
